package sunyu.tools.hbase.mapper;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import sunyu.tools.hbase.mapper.pojo.Family;
import sunyu.tools.hbase.mapper.pojo.Row;

import java.util.*;


/**
 * 将hbase查询的结果集转换成java bean
 *
 * @author 孙宇
 */
public class HbaseRowMapper {

    /**
     * result转换row对象
     *
     * @param result             hbase查询出的结果
     * @param returnRowKeyOnly   是否只看rowkey
     * @param showHistoryVersion 是否转换历史版本数据(需要scan中scan.setMaxVersions(Integer.MAX_VALUE);或者get.setMaxVersions(Integer.MAX_VALUE);)
     * @return
     */
    public static Row mapRow(Result result,
                             Boolean returnRowKeyOnly,
                             Boolean showHistoryVersion) {
        List<Row> rows = mapRow(Arrays.asList(result), returnRowKeyOnly, showHistoryVersion);
        if (rows != null && rows.size() > 0) {
            return rows.get(0);
        }
        return null;
    }

    /**
     * result转换row对象
     *
     * @param results            hbase查询出的结果
     * @param returnRowKeyOnly   是否只看rowkey
     * @param showHistoryVersion 是否转换历史版本数据(需要scan中scan.setMaxVersions(Integer.MAX_VALUE);或者get.setMaxVersions(Integer.MAX_VALUE);)
     * @return
     */
    public static List<Row> mapRow(List<Result> results,
                                   Boolean returnRowKeyOnly,
                                   Boolean showHistoryVersion) {
        if (results != null && results.size() > 0) {
            List<Row> rows = new ArrayList<>();
            for (Result result : results) {
                if (result == null || result.getRow() == null) {
                    return null;
                }
                Row row = new Row();// 一行
                row.setRowName(Bytes.toString(result.getRow()));
                if (returnRowKeyOnly == null || returnRowKeyOnly == false) {//如果没传只返回rowkey的参数，或者这个参数是false
                    Map<String, Family> familys = row.getFamilys();// 一行里的family集合
                    for (Cell cell : result.rawCells()) {
                        String familyName = Bytes.toString(CellUtil.cloneFamily(cell));// 获取当前列的familyName
                        String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));// 列名
                        String value = Bytes.toString(CellUtil.cloneValue(cell));// 值

                        Family family = familys.get(familyName);
                        if (family == null) {
                            family = new Family();
                            familys.put(familyName, family);
                        }

                        Map<String, Object> columns = family.getColumns();// 当前family的所有列
                        columns.put(columnName, value);
                    }

                    //多版本的时候
                    if (showHistoryVersion != null && showHistoryVersion == true) {
                        for (Cell c : result.listCells()) {
                            String familyName = Bytes.toString(CellUtil.cloneFamily(c));// 获取当前列的familyName
                            String columnName = Bytes.toString(CellUtil.cloneQualifier(c));// 列名
                            String value = Bytes.toString(CellUtil.cloneValue(c));// 值
                            long timestamp = c.getTimestamp();//插入时间

                            Family family = familys.get(familyName);
                            Map<String, Object> historyColumns = family.getHistoryColumns();

                            List columnVersionList = (List) historyColumns.get(columnName);//当前列的多版本map
                            if (columnVersionList == null) {
                                columnVersionList = new ArrayList();
                                historyColumns.put(columnName, columnVersionList);
                            }

                            Map<String, Object> currentColumn = new HashMap<String, Object>();
                            currentColumn.put("value", value);
                            currentColumn.put("timestamp", timestamp);
                            columnVersionList.add(currentColumn);
                        }
                    }
                }
                rows.add(row);
            }
            return rows;
        }
        return null;
    }

}